package com.dcy.daily.skill.rabbitMQ.official.rpc;

import com.dcy.daily.skill.rabbitMQ.official.utils.MQConnectUtil;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeoutException;

/**
 * 使用MQ来实现RPC功能
 * RPC客户端，发送给服务端一个数字，从服务端得到该数字的斐波那契值
 *
 * @author duanchunyu
 * @version 1.0.0
 * @createTime 2021年07月21日 10:48
 */
public class RPCClient implements AutoCloseable {
    private Connection connection;
    private String requestQueueName = "rpc_queue";
    private Channel channel;

    public RPCClient() throws IOException, TimeoutException {
        connection = MQConnectUtil.getConnection();
        channel = connection.createChannel();
    }

    public static void main(String[] args) {
        try (RPCClient rpcClient = new RPCClient()) {
            for (int i = 0; i < 32; i++) {
                String i_str = Integer.toString(i);
                System.out.println(" [x] Requesting fib(" + i_str + ")");
                String response = rpcClient.call(i_str);
                System.out.println(" [.] Got '" + response + "'");
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 发送RPC请求
     * @param n 要计算的斐波那契数
     * @return
     * @throws IOException
     * @throws InterruptedException
     */
    public String call(String n) throws IOException, InterruptedException {
        final String corrId = UUID.randomUUID().toString();
        String replyQueueName = channel.queueDeclare().getQueue();
        AMQP.BasicProperties prop = new AMQP.BasicProperties()
                .builder()
                .correlationId(corrId)
                .replyTo(replyQueueName)
                .build();
        channel.basicPublish("",requestQueueName,prop,n.getBytes(StandardCharsets.UTF_8));

        final BlockingQueue<String> response = new ArrayBlockingQueue<>(1);
        String ctag = channel.basicConsume(replyQueueName, true, (consumerTag, delivery) -> {
            if (delivery.getProperties().getCorrelationId().equals(corrId)) {
                response.offer(new String(delivery.getBody(), StandardCharsets.UTF_8));
            }
        }, consumerTag -> {
        });

        String result = response.take();
        channel.basicCancel(ctag);
        return result;
    }

    @Override
    public void close() throws Exception {
        connection.close();
    }
}
